/*
 * (C) 2009, Elephant Seven Hamburg Gmbh, 
 * Author: Detlev Wiese
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not 
 * use this file except in compliance with the License. You may obtain a copy 
 * of the License at
 *  
 *     http://www.apache.org/licenses/LICENSE-2.0 
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.se and its accompanying notice file is
 * included with the distribution. 
 */ 
package com.e7.streams;

import java.util.NoSuchElementException;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * <P>This class represents streams which are very like  
 * MappedStream-instances (see {@link MappedStream}). Instead 
 * of sequentially applying the function given, streams of 
 * this class evaluate the function on the wrapped stream in 
 * parallel for more than one element. Still elements of the this 
 * stream can be read sequentially only.</P>
 * <P>Parallel evaluation is done using a thread pool, which 
 * can be provided on instance-creation or is created in the 
 * constructors. To work properly the {@link ThreadPoolExecutor} 
 * must have a {@link BlockingQueue} which has no size limitation.</P>
 * <P>This class implements two different behaviors: eager or 
 * not. Eager means, the underlying stream is mapped as fast as 
 * possible by running ahead through the underlying stream until 
 * its end. This behavior will use memory to queue the results.    
 * When behaving not eagerly, there will be always parallel 
 * evaluation, but the result queue is limited in length to the 
 * size of the thread pool. That means that the evaluation will 
 * stop, when results are not consumed.</P> 
 * <P>You might want to use the non eager behavior in cases where 
 * you have a very long underlying stream or very memory consuming 
 * results. Never use eager behavior on infinite streams.</P>
 * @param <F> The type of the function's result
 * @param <E> The type of the wrapped stream
 */
public class ThreadedStream<F, E> extends AbstractStream<F> {	
	
	protected final MappingFunction<F, E> function;
	protected final Stream<E> stream;
	protected final ExecutorService executorService; 
	protected final LinkedBlockingDeque<Future<F>> futuresQueue; 
	protected boolean eager;
  protected final boolean shutDownOnFinalize; 
	protected final int maxThreads; 
	
	/**
	 * Create an instance! 
	 * @param maxThreads - the number of concurrently running 
	 * function evaluations
	 * @param function - the function which is evaluated for 
	 * all elements concurrently
	 * @param stream - the underlying stream providing elements 
	 * to be mapped by the function given.
	 */
	public ThreadedStream(
				int maxThreads, 
				MappingFunction<F, E> function,
				Stream<E> stream) 
	{
		if (maxThreads < 1)
			throw new IllegalArgumentException();
		
		this.function = function;
		this.stream = stream;
		this.eager = false;
		this.executorService = 
			new ThreadPoolExecutor(
					maxThreads, 
					maxThreads, 
					0, 
					TimeUnit.MILLISECONDS, 
					new LinkedBlockingQueue<Runnable>());
		this.maxThreads = maxThreads;
		this.shutDownOnFinalize = true; // we have created a new thread pool..
		this.futuresQueue = new LinkedBlockingDeque<Future<F>>();
	}

	/**
	 * See {@link ThreadedStream#ThreadedStream(int, MappingFunction, Stream)}
	 * @param eager - true, means you get eager behavior
	 */
	public ThreadedStream(
			int maxThreads, 
			boolean eager, 
			MappingFunction<F, E> function,
			Stream<E> stream) { 
		this(maxThreads, function, stream); 
		this.eager = eager; 
	} 
	
	/**
	 * Create an instance! 
	 * @param function - the function which is evaluated for 
	 * all elements concurrently
	 * @param stream - the underlying stream providing elements 
	 * to be mapped by the function given.
	 * @param executionService - a thread pool with {@link BlockingQueue}
	 * which must not have a capacity limit. 
	 */
	public ThreadedStream( 
	    int maxThreads, 
			MappingFunction<F, E> function,
			Stream<E> stream, 
			ExecutorService executorService) { 
	  if (null == executorService)
	    throw new IllegalArgumentException("executorService is null");
	  
	  this.maxThreads = maxThreads;
		this.executorService = executorService;
		this.shutDownOnFinalize = false;
		this.function = function; 
		this.stream = stream; 
		this.eager = false; 
		this.futuresQueue = new LinkedBlockingDeque<Future<F>>();
	} 
	
	/**
	 * See {@link 
	 * ThreadedStream#ThreadedStream(MappingFunction, Stream, ThreadPoolExecutor)}
	 * @param eager - true, means you get eager behavior
	 */
	public ThreadedStream( 
	    int maxThreads, 
			boolean eager,
			MappingFunction<F, E> function,
			Stream<E> stream, 
			ExecutorService executorService) { 
		this(maxThreads, function, stream, executorService);
		this.eager = eager;
	} 
	
	public boolean isEager() {
		return eager;
	}

	public void setEager(boolean beEager) {
		this.eager = beEager;
	}

	/**
	 * The receiver will create an instance implementing the 
	 * {@link Callable} interface. This instance can be used with 
	 * Executor-Service instances. 
	 * can be processed by  
	 * @param function the callable will apply this function
	 * @param element the function will be applied to this element. 
	 * @return the new created Callable-instance
	 */
	protected Callable<F> createCallable(
								final MappingFunction<F, E> function, 
								final E element) 
	{
		return new Callable<F>() {
			public F call() throws Exception {
				return function.value(element);
			}
		};
	}

	/**
	 * This method starts new threads, it is called whenever the 
	 * method ThreadStream#next() is called, and, thus an element
	 * is about to be delivered.  
	 */
	protected void startThreads() {
		while ((eager || 
					futuresQueue.size() < maxThreads) 
					&& stream.hasNext()) 
			futuresQueue.add(executorService.submit(
					this.createCallable(function, stream.next())));
	}
	
	/**
	 * <P>This method will not start new threads, it uses the underlying 
	 * stream to decide whether there will be more elements to read.</P>
	 */
	public boolean hasNext() {
		return !futuresQueue.isEmpty() || stream.hasNext();
	}
	
	/**
	 * <P>Return the next element!</P>
	 * <P>As all elements delivered will be the result of 
	 * executing a separate thread, there might be cases 
	 * where the thread execution did not yield a valid result 
	 * but has thrown an exception. Those exceptions will turn 
	 * up here as {@link NoSuchElementException}. The exception 
	 * raised will have its cause initialized to allow access to 
	 * the causing exception. </P> 
	 */
	public F next() {
		this.startThreads();
		
		if (futuresQueue.isEmpty()) 
			throw new NoSuchElementException();
		
		try {
			Future<F> future = futuresQueue.pollFirst();
			return future.get();
		} catch (Exception e) {
			NoSuchElementException x = 
				new NoSuchElementException("error while executing thread"); 
			x.initCause(e);
			throw x; 
		}
	}
	
	/**
	 * The receiver returns the current number of futures it expects 
	 * to be computed by the execution service.
	 * @return
	 */
	public int queueSize() {
	  return futuresQueue.size();
	}
	
	/**
	 * We must kill any running threads here! Their result is not 
	 * reachable.
	 */
	protected void finalize() throws Throwable {
		if (shutDownOnFinalize) 
			executorService.shutdown();
		super.finalize();
	}
}
